package cn.tedu.charging.order.mqtt;

import cn.tedu.charging.common.constant.MqttDestinationConst;
import cn.tedu.charging.common.pojo.message.ChargingCheckResultMessage;
import cn.tedu.charging.common.pojo.message.ChargingProgressMsg;
import cn.tedu.charging.order.service.OrderConsumerService;
import cn.tedu.charging.order.service.OrderConsumerServiceImpl;
import cn.tedu.charging.order.service.OrderService;
import com.alibaba.fastjson2.JSON;
import lombok.extern.slf4j.Slf4j;
import org.eclipse.paho.client.mqttv3.IMqttDeliveryToken;
import org.eclipse.paho.client.mqttv3.MqttCallbackExtended;
import org.eclipse.paho.client.mqttv3.MqttClient;
import org.eclipse.paho.client.mqttv3.MqttMessage;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.stereotype.Component;

import java.nio.charset.StandardCharsets;
import java.time.Duration;
import java.util.concurrent.TimeUnit;

/**
 * 处理emqx的主题消息的唯一消费者
 * 订单中所有也业务消费都由这个类实现 需要最终处理的消费逻辑有2个
 * 1. 接收设备反馈信息
 * TODO 2. 接收设备充电进度信息
 */
@Slf4j
public class MqttConsumer implements MqttCallbackExtended {
    private MqttClient mqttClient;
    private MqttContextHook mqttContextHook;

    public MqttConsumer(MqttClient mqttClient, MqttContextHook mqttContextHook) {
        this.mqttClient = mqttClient;
        this.mqttContextHook = mqttContextHook;
    }

    public MqttConsumer() {
    }

    @Override
    public void connectComplete(boolean reconnect, String serverURI) {
        //使用连接对象监听主题 接收设备反馈
        try {
            log.debug("mqtt开始订阅主题...");
            mqttClient.subscribe(MqttDestinationConst.GUN_CHECK_RESULT_TOPIC, 1);
            mqttClient.subscribe(MqttDestinationConst.CHARGING_PROGRESS_TOPIC, 1);
            log.debug("mqtt已经订阅主题:{}", MqttDestinationConst.GUN_CHECK_RESULT_TOPIC);
            log.debug("mqtt已经订阅主题:{}", MqttDestinationConst.CHARGING_PROGRESS_TOPIC);
        } catch (Exception e) {
            log.error("主题订阅异常", e);
        }
    }

    @Override
    public void messageArrived(String topic, MqttMessage message) throws Exception {
        if (topic.equals(MqttDestinationConst.GUN_CHECK_RESULT_TOPIC)) {
            //主题 设备自检有反馈
            handleCheckResult(topic, message);
        } else if (topic.equals(MqttDestinationConst.CHARGING_PROGRESS_TOPIC)) {
            //充电进度同步
            handleChargingProgress(topic, message);
        }
    }

    private void handleChargingProgress(String topic, MqttMessage message) {
        String jsonMsg = new String(message.getPayload(), StandardCharsets.UTF_8);
        log.debug("主题:{},消息:{}", topic, jsonMsg, StandardCharsets.UTF_8);
        // 解析，将json转化成接口消息数据
        if(jsonMsg == null || "".equals(jsonMsg.trim())){
            log.debug("消息为空");
            return;
        }
        ChargingProgressMsg msg = JSON.parseObject(jsonMsg, ChargingProgressMsg.class);
        mqttContextHook.getOrderConsumerService().handleChargingProgress(msg);
    }

    private void handleCheckResult(String topic, MqttMessage message) {
        //先不考虑消费逻辑如何执行 先检查流程跑通
        log.debug("主题:{},消息:{}", topic, new String(message.getPayload(), StandardCharsets.UTF_8));
        //1.解析消息 获得反序列化的对象
        ChargingCheckResultMessage msg = null;
        String strMsg = new String(message.getPayload(), StandardCharsets.UTF_8);
        msg = JSON.parseObject(strMsg, ChargingCheckResultMessage.class);
        //2.mqttContext调用获取get orderService 调用消费
        //2.1抢锁 setnx key(orderNo) "" EX 3
        RedisTemplate redisTemplate = mqttContextHook.getRedisTemplate();
        String lockKey = "charging:device:result:" + msg.getOrderNo();
        Boolean getKey = redisTemplate.boundValueOps(lockKey).setIfAbsent("", 3, TimeUnit.SECONDS);
        try {
            if (getKey) {
                log.debug("获取锁成功,开始处理消息");
                mqttContextHook.getOrderConsumerService().handleCheckResult(msg);
            } else {
                log.debug("获取锁失败,结束消息消费");
            }
        } catch (Exception e) {
            log.error("处理消息异常", e);
        } finally {
            redisTemplate.delete(lockKey);
        }
    }

    @Override
    public void connectionLost(Throwable cause) {
        log.error("mqtt连接断开", cause);
    }

    /**
     * 充电桩项目可以使用deliverComplete 打印生产发送的结果 qos=1 | 2生效 0 不会调用
     */
    @Override
    public void deliveryComplete(IMqttDeliveryToken token) {
        log.debug("消息发送成功,消息id:{}", token.getMessageId());
    }

    private <T> T deserialize(byte[] bytes, Class<T> clazz) {
        return JSON.parseObject(bytes, clazz);
    }
}
